package com.gator.rocketmq.producer;

import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author PABLO
 * @Date 2022/6/8 12:00
 * @Desc 分布式事务
 */
public class TransactionProducer {
    public static void main(String[] args) throws MQClientException {

        //监听事务状态
        TransactionListener listener = new TransactionListener() {
            //2.如发送成功，执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                try {
                    //执行本地事务
                    System.out.println("执行本地事务");
                    if (true) {
                        //如成功
                        return LocalTransactionState.COMMIT_MESSAGE;
                    }
                } catch (Exception e) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                //如操作时间很长，为了不阻塞，返回该状态，则依赖于后续MQ的回查机制checkLocalTransaction
                return LocalTransactionState.UNKNOW;

            }

            //MQ检查生产者本地事务
            //默认60S一次
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                try {
                    //检查本地事务是否成功
                    if (true) {
                        //如成功
                        return LocalTransactionState.COMMIT_MESSAGE;
                    }
                } catch (Exception e) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                //如操作时间很长，为了不阻塞，返回该状态，则依赖于后续MQ的回查机制checkLocalTransaction
                return LocalTransactionState.UNKNOW;
            }
        };

        TransactionMQProducer producer = new TransactionMQProducer("transaction_pablo_group");

        producer.setNamesrvAddr("localhost:9876");

        producer.setUnitName("456");

        //设置回查线程池
        producer.setExecutorService(new ThreadPoolExecutor(2, 5, 100, TimeUnit.MINUTES,
                new ArrayBlockingQueue<Runnable>(10), new DefaultThreadFactory("pablo_transactional"),
                new ThreadPoolExecutor.CallerRunsPolicy()));
        //生产者监听器
        producer.setTransactionListener(listener);

        producer.start();

        try {
            //1.半事务发送
            Message message = new Message("pablo_transaction_topic", "a-->b".getBytes(StandardCharsets.UTF_8));

            TransactionSendResult result = producer.sendMessageInTransaction(message, null);
            SendStatus sendStatus = result.getSendStatus();
            if (sendStatus.equals(SendStatus.SEND_OK)) {
                System.out.println("发送半事务状态成功");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}

@Slf4j
class TransactionConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Transaction_consumer_group");

        consumer.setNamesrvAddr("localhost:9876");

        consumer.subscribe("pablo_transaction_topic", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                try {
                    for (MessageExt item : msgs
                    ) {
                        String topic = item.getTopic();
                        String msg = null;
                        try {
                            msg = new String(item.getBody(), "UTF-8");
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                        String tags = item.getTags();
                        log.info("a-->b{}，tag是：{}，topic是：{}", msg, tags, topic);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                } catch (Exception e) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return null;
            }
        });

        consumer.start();
    }
}
